package com.own.business.sync.trigger.polling.limit.core;

import com.own.business.sync.trigger.base.bak.util.SyncTriggerBakUtil;
import com.own.business.sync.trigger.base.config.SyncTriggerPollingCondition;
import com.own.business.sync.trigger.base.method.ResolveMethodSpringHook;
import com.own.business.sync.trigger.base.property.SyncTriggerProperty;
import com.own.business.sync.trigger.base.service.SyncTriggerService;
import com.own.component.common.util.executor.ExecutorUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Conditional;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashSet;

/**
 * SyncTriggerPollingLimitCoreTask
 * <p>
 * 启动完成后开始执行
 *
 * @author chenxueli
 * @date 2024-01-08 15:48:43
 */
@Slf4j
@Component
@Conditional(SyncTriggerPollingCondition.class)
public class SyncTriggerPollingLimitCoreTask {

    private final Object lock = new Object();
    @Resource
    private SyncTriggerProperty syncTriggerProperty;
    @Resource
    private SyncTriggerService syncTriggerService;
    @Resource
    private SyncTriggerBakUtil syncTriggerBakUtil;
    @Resource
    private ResolveMethodSpringHook resolveMethodSpringHook;
    @Resource
    private ExecutorUtil executorUtil;

    @PostConstruct
    public void init() {
        executorUtil.execute(this::polling);
    }

    /**
     * 轮询信息
     */
    @SuppressWarnings("all")
    private void polling() {
        var service = ExecutorUtil.newVirtualThreadPerTaskExecutor();
        var max = syncTriggerProperty.getPollingTimeList().length - 1;
        // 采用while true的方式可以避免使用递归方式造成的栈溢出的问题
        var c = 0;
        while (true) {
            synchronized (lock) {
                // 查询当前备份库的数量信息
                var list = syncTriggerService.listForLimit();
                if (!list.isEmpty()) {
                    var updateList = new HashSet<String>();
                    // 数据处理，如果同一条数据的修改出现过多次，只需要提交一次即可
                    list.forEach(data -> {
                        // 获取到最新的数据去执行信息
                        var method = resolveMethodSpringHook.getMethod(data.getModuleName());
                        if (method != null) {
                            var operation = resolveMethodSpringHook.getMethod(data.getType(), method);
                            var key = data.getModuleName() + ":" + data.getType() + ":" + data.getTargetId();
                            if (!updateList.contains(key)) {
                                updateList.add(key);
                                // 使用虚拟线程进行消费
                                service.execute(() -> {
                                    try {
                                        operation.accept(data);
                                    } catch (Exception e) {
                                        log.debug("处理异常", e);
                                        syncTriggerBakUtil.save(data, e.getMessage());
                                    }
                                });
                            }
                            syncTriggerService.deleteById(data.getId());
                        } else {
                            syncTriggerBakUtil.save(data, "未找到处理方法");
                            syncTriggerService.deleteById(data.getId());
                        }
                        log.debug("处理完成:{},{},{}", data.getModuleName(), data.getType(), data.getTargetId());
                    });
                    log.info("完成处理{}条数据，开始进入下一个循环", list.size());
                    c = 0;
                } else {
                    var index = Math.min(c, max);
                    var timeout = syncTriggerProperty.getPollingTimeList()[index];
                    log.info("未查询到调整的数据，开始休眠" + timeout + "ms");
                    try {
                        lock.wait(timeout);
                        c++;
                    } catch (InterruptedException e) {
                        log.error("等待异常", e);
                    }
                }
            }
        }
    }

}
